package ex.tableapi.aggregation;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

public class ExGroupByWindowAggregation extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExGroupByWindowAggregation demo = new ExGroupByWindowAggregation();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        Table result = demo.exGroupByWindowAggregation(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }




    private Table exGroupByWindowAggregation(Table orders) {
        // a,rowtime分组，统计
        Table result = orders
                .window(Tumble.over(lit(35).minutes()).on($("rowtime")).as("w")) //在rowtime 定义5分钟的滚动窗口并给窗口起一个别名w
                .groupBy($("a"), $("w")) // group by key and window
                // access window properties and aggregate
                .select(
                        $("a"),
                        $("w").start(),
                        $("w").end(),
                        $("w").rowtime(),
                        $("b").sum().as("amount")
                );
//        1> +I[liji, 2024-01-24T18:45, 2024-01-24T19:20, 2024-01-24T19:19:59.999, 1100]  含 500、600
//        1> +I[liji, 2024-01-24T19:20, 2024-01-24T19:55, 2024-01-24T19:54:59.999, 700]
//        1> +I[liji, 2024-01-25T18:40, 2024-01-25T19:15, 2024-01-25T19:14:59.999, 1700]
        return result;
    }



}